package com.edata.bigdata.basic;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.SparkSessionBuilder;
import shaded.parquet.org.apache.thrift.protocol.TCompactProtocol;

import java.util.HashMap;
import java.util.Map;

public class Manager {

    public String APPNAME = "EDATA";
    /*
     *1.local：本地模式，单线程，没有任何并行计算
     *2.local[k]：本地模式，指定使用多少个线程并行计算，k即使用k个Worker线程
     *3.local[*]：本地模式，按照CPU最多cores设置线程数
     *4.spark://[host]:[port]，Standalone模式，Spark自己负责资源的管理调度。此时master通常就一个。
     *5.mesos://[host]:[port]：使用mesos来管理资源调度
     *6.yarn 采用yarn来管理资源调度，进一步可分成 cluster 和 client，由deploy-mode参数指定
     *   6.1.cluster 生产环境常用的模式，所有的资源调度和计算都在集群环境上运行
     *   6.2.client Spark Driver和ApplicationMaster进程均在本机运行，而计算任务在cluster
     *   此时MASTER设置为yarn，提交作业的命令如下：
     *   $SPARK_HOME/bin/spark-submit --master yarn --deploy-mode cluster your-spark-app.jar
     *7.Spark standalone模式下
     *   7.1.当使用 idea 远程连接 spark 引擎调试代码，此时 driver 在 client 端，即 idea 所在的 ip 节点，此时
     *       需要指定idea所在所在的 ip 和 port，以提供 worker 与 driver 通信。
     *       例如：CONFIG.put("spark.driver.host","192.168.36.1")，CONFIG.put("spark.driver.port","7076")
     *
     * */
    public String MASTER = "local[*]";
    public SparkSession SPARKSESSION = null;
    public Map<String, String> CONFIG = new HashMap<String, String>();

    public StreamExecutionEnvironment flinkEnv = null;
    public Configuration flinkConf = new Configuration();


    public void createSparkSession() {
        SparkConf sparkConf = new SparkConf();
        SparkSessionBuilder ssb = SparkSession.builder().appName(APPNAME).master(MASTER);
        if (!CONFIG.isEmpty()) {
            for (String key : CONFIG.keySet()) {
                if (key.equals("spark.serializer")) {
                    sparkConf.registerKryoClasses(new Class[]{
                            TCompactProtocol.class
                    });
                } else {
                    sparkConf.set(key, CONFIG.get(key));
                }
            }
        }
        ssb.config(sparkConf);
        SPARKSESSION = ssb.getOrCreate();
    }

    public void createFlinkEnvironment(){
        flinkEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        flinkEnv.configure(flinkConf);
    }

    public void stop(){
        if(SPARKSESSION != null){
            SPARKSESSION.stop();
        }
    }

    public Manager() {
    }
}
